package com.bw.gmall.realtime.app.zyjdwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Zyj_DwdTradeCartAdd {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableResult sql = tenv.executeSql("CREATE TABLE topic_db ( " +
                "  `database` STRING, " +
                "  `table` STRING, " +
                "  `type` STRING, " +
                "  `data` MAP<STRING,STRING>, " +
                "  `old` MAP<STRING,STRING>, " +
                "  `pt` AS PROCTIME() " +
                ") " +
                        "WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'topic_db',\n" +
                        "  'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                        "  'properties.group.id' = 'topic_db1234',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'format' = 'json'\n" +
                        ")"
                );
//        tenv.sqlQuery("select * from topic_db").execute().print();
        Table cartAddTable = tenv.sqlQuery("" +
                "select " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    `data`['cart_price'] cart_price, " +
                "    if(`type`='insert',`data`['sku_num'],cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)) sku_num, " +
                "    `data`['sku_name'] sku_name, " +
                "    `data`['is_checked'] is_checked, " +
                "    `data`['create_time'] create_time, " +
                "    `data`['operate_time'] operate_time, " +
                "    `data`['is_ordered'] is_ordered, " +
                "    `data`['order_time'] order_time, " +
                "    `data`['source_type'] source_type, " +
                "    `data`['source_id'] source_id, " +
                "    pt " +
                "from topic_db " +
                "where `database` = 'gmall' " +
                "and `table` = 'cart_info' " +
                "and (`type` = 'insert' " +
                "or (`type` = 'update'  " +
                "    and  " +
                "    `old`['sku_num'] is not null  " +
                "    and  " +
                "    cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)))");
        tenv.createTemporaryView("cart_info_table", cartAddTable);
//        tenv.sqlQuery("select  *  from cart_info_table ").execute().print();
        tenv.executeSql( "CREATE TABLE base_dic (\n" +
               "`dic_code` string, " +
               "`dic_name` string, " +
               "`parent_code` string, " +
               "`create_time` timestamp, " +
               "`operate_time` timestamp, " +
                "primary key(`dic_code`) not enforced " +
                ") WITH (\n" +
                "   'connector' = 'jdbc',\n" +
                "   'url' = 'jdbc:mysql://hadoop102:3306/gmall',\n" +
                "   'table-name' = 'base_dic' ,\n" +
                "   'lookup.cache.max-rows' = '10',\n" +
                "   'lookup.cache.ttl' = '1 hour' ,\n" +
                "   'username' = 'root',\n" +
                "   'password' = 'root',\n" +
                "'driver' = 'com.mysql.cj.jdbc.Driver' " +
                ")");
//        tenv.sqlQuery("select * from base_dic").execute().print();
        Table table = tenv.sqlQuery("select " +
                "    ci.id, " +
                "    ci.user_id, " +
                "    ci.sku_id, " +
                "    ci.cart_price, " +
                "    ci.sku_num, " +
                "    ci.sku_name, " +
                "    ci.is_checked, " +
                "    ci.create_time, " +
                "    ci.operate_time, " +
                "    ci.is_ordered, " +
                "    ci.order_time, " +
                "    ci.source_type source_type_id, " +
                "    dic.dic_name source_type_name, " +
                "    ci.source_id " +
                "from cart_info_table ci " +
                "join base_dic FOR SYSTEM_TIME AS OF ci.pt as dic " +
                "on ci.source_type = dic.dic_code");
        tenv.createTemporaryView("cart_add_dic_table", table);
        table.execute().print();
        tenv.executeSql("CREATE TABLE dwd_cart_add (\n" +
                "    id STRING COMMENT '购物车ID',\n" +
                "    user_id STRING COMMENT '用户ID',\n" +
                "    sku_id STRING COMMENT '商品SKU ID',\n" +
                "    cart_price DECIMAL(10, 2) COMMENT '加入购物车时的商品价格',\n" +
                "    sku_num INT COMMENT '商品数量',\n" +
                "    sku_name STRING COMMENT '商品名称',\n" +
                "    is_checked TINYINT COMMENT '是否勾选（0-未勾选，1-勾选）',\n" +
                "    create_time TIMESTAMP(3) COMMENT '创建时间',\n" +
                "    operate_time TIMESTAMP(3) COMMENT '操作时间',\n" +
                "    is_ordered TINYINT COMMENT '是否已下单（0-未下单，1-已下单）',\n" +
                "    order_time TIMESTAMP(3) COMMENT '下单时间',\n" +
                "    source_type_id STRING COMMENT '来源类型ID（字典表code）',\n" +
                "    source_type_name STRING COMMENT '来源类型名称（字典表name）',\n" +
                "    source_id STRING COMMENT '来源ID（如：商品详情页、推荐列表等）'\n" +
                ") "+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cart_add"));
        tenv.executeSql("insert into dwd_cart_add select * from cart_add_dic_table");
        env.execute("DwdTradeCartAdd");
//        env.execute();
    }
}
